
#pragma once
#include"epoll.h"
#include"EventLoop.h"
#include <condition_variable>
#include <mutex>

class Connection{
    private:
        using callback_func=std::function<void(void)>;
        //DISCONECTED -- 连接关闭状态；   CONNECTING -- 连接建立成功-待处理状态
        //CONNECTED -- 连接建立完成，各种设置已完成，可以通信的状态；  DISCONNECTING -- 待关闭状态
        typedef enum { DISCONNECTED, CONNECTING, CONNECTED, DISCONNECTING}ConnStatu;
        using callback=std::function<void()>;
        int _conn_id;//不可以用_sockfd作为timer_wheel的id，因为fd会被重复利用
        EventLoop* _loop;
        Socket _socket;
        Channel _channel;
        Buffer _in_buffer,_out_buffer;
        ConnStatu _status;
        bool _enable_inactive_release;
        Any _any;

        //这四个是给使用者提供的
        using ConnectedCallback = std::function<void( Connection*)>;//连接建立完成后应该干什么
        using MessageCallback = std::function<void( Connection*, Buffer *)>;//收到消息后应该干什么
        using ClosedCallback = std::function<void( Connection*)>;//连接关闭后应该干什么
        using AnyEventCallback = std::function<void( Connection*)>;//任何事件发生时应该干什么
        ConnectedCallback _connected_callback;
        MessageCallback _message_callback;
        ClosedCallback _closed_callback,_server_closed_callback;
        AnyEventCallback _event_callback;

        //处理读事件
        void HandleReadEvenet(){
            //std::cout<<"HandleReadEvenet"<<__LINE__<<std::endl;
            int n=_socket.Recv(_in_buffer,0);//
            if(n<0){
                perror("recv error");
                Close();
            }
            _message_callback(this,&_in_buffer);
        }
        void HandleWriteEvent(){
            string msg=_out_buffer.ReadAndPop();
            bool ret=_socket.Send(msg);
            if(!ret){
                HandleCloseEvent();
            }
            //没有数据发送就关闭写事件监听
            if(_out_buffer.GetFreeSpace()<=0){
                _channel.DisableWriteEvent();
            }
        }
        //处理连接关闭事件,客户端关闭
        void HandleCloseEvent(){
            if(_in_buffer.GetSize()>0){
                _message_callback(this,&_in_buffer);
                return;
            }
            //不能直接关闭，因为还需要处理读事件
            if(_out_buffer.GetSize()>0){
                if(_channel.WriteEanble()){
                    _channel.EnableWriteEvent();
                }
                return;
            }
            //std::cout<<"HandleCloseEvent"<<__LINE__<<std::endl;
            CloseInloop();
        }
        void HandleErrorEvent(){
            perror("HandleErrorEvent");std::cout<<__LINE__<<std::endl;
            CloseInloop();
        }
        void HandleEvent(){
            if(_status==DISCONNECTED||_status==DISCONNECTING){
                return;
            }
            //刷新定时关闭任务
            if(_enable_inactive_release){
                _loop->TimerRefresh(_conn_id);
            }
            if(_event_callback){
                _event_callback(this);
            }
        }
        //连接建立成功
        void EstablishedInLoop(){
            assert(_status==CONNECTING);
            _status=CONNECTED;
            //启动读监控
            LOG(ly::LogLevel::DEBUG)<<"EstablishedInLoop:"<<_channel.Fd();
            _channel.EnableReadEvent();//这里把Channel添加进入EventLoop中了
            if(_connected_callback){
                _connected_callback(this);
            }
        }
        //真正关闭连接
        void CloseInloop(){
            if(_status==DISCONNECTING){
                return;
            }
            //更改状态
            _status=DISCONNECTING;
            _channel.Remove();
            //关闭socket
            _socket.Close();
            if(_loop->HasTimerTask(_conn_id)){
                _loop->TimerCancel(_conn_id);
            }
            //调用关闭回调
            if(_closed_callback)
                _closed_callback(this);
            //删除自己,需要在管理层面删除
            if(_server_closed_callback)
                _server_closed_callback(this);
            _status==DISCONNECTED;
        }
        //将数据写入到缓冲区
        void SendInLoop(const string& _msg){
            _out_buffer.Write(_msg);
            _channel.EnableWriteEvent();
            if(!_channel.WriteEanble()){
                _channel.EnableWriteEvent();
            }
        }
        //设置定时任务
        void SetTimerTaskInloop(int timeout){
            if(_loop->HasTimerTask(_conn_id)){
                _loop->TimerRefresh(_conn_id);
                return;
            }
            _loop->AddTimerTask(_conn_id,timeout,std::bind(&Connection::Close,this));
            _enable_inactive_release=true;
        }
        void CancelInactiveReleaseInLoop(){
            //取消定时任务
            if(_loop->HasTimerTask(_conn_id)){
                _loop->TimerCancel(_conn_id);
            }
            _enable_inactive_release=false;
        }
        void UpgradeInLoop(const Any &context, 
                    const ConnectedCallback &conn, 
                    const MessageCallback &msg, 
                    const ClosedCallback &closed, 
                    const AnyEventCallback &event) {
            _any = context;
            _connected_callback = conn;
            _message_callback = msg;
            _closed_callback = closed;
            _event_callback = event;
        }
    public:
        Connection(EventLoop* loop,int conn_id,int sockfd):_loop(loop),_conn_id(conn_id),
                    _socket(sockfd),_channel(sockfd,loop),_status(CONNECTING),_enable_inactive_release(false){
            
            //设置Channel回调
            _channel.SetReadCallback(std::bind(&Connection::HandleReadEvenet,this));
            _channel.SetWriteCallback(std::bind(&Connection::HandleWriteEvent,this));
            _channel.SetCloseCallback(std::bind(&Connection::HandleCloseEvent,this));
            _channel.SetEventCallback(std::bind(&Connection::HandleEvent,this));
            _channel.SetErrorCallback(std::bind(&Connection::HandleErrorEvent,this));
        }
        int Fd() {
            return _channel.Fd();
        }
        int Id() {
            return _conn_id;
        }
        // using ConnectedCallback = std::function<void(const Connection*)>;//连接建立完成后应该干什么
        // using MessageCallback = std::function<void(const Connection*, Buffer *)>;//收到消息后应该干什么
        // using ClosedCallback = std::function<void(const Connection*)>;//连接关闭后应该干什么
        // using AnyEventCallback = std::function<void(const Connection*)>;//任何事件发生时应该干什么
        void SetConnectedCallback(ConnectedCallback& callback){
            _connected_callback=callback;
        }
        void SetMessageCallback(MessageCallback& callback){
            _message_callback=callback;
        }
        void SetCloseCallback( ClosedCallback& callback){
            _closed_callback=callback;
        }
        void SetAnyEventCallback( AnyEventCallback& callback){
            _event_callback=callback;
        }
        void SetServerClosedCallback( const ClosedCallback& callback){
            _server_closed_callback=callback;
        }
        //连接建立成功
        void Established(){
            _loop->PushTask(std::bind(&Connection::EstablishedInLoop,this));
        }
        //启动定时销毁任务
        void EnableInactiveRelease(int timeout){
            _loop->PushTask(std::bind(&Connection::SetTimerTaskInloop,this,timeout));
        }
        //取消定时销毁任务
        void CancelInactiveRelease(){
            _loop->PushTask(std::bind(&Connection::CancelInactiveReleaseInLoop,this));
        }
        //将信息写入缓冲区
        void Send(const char* data,int size=0) {
            string msg(data,size);
            _loop->PushTask(std::bind(&Connection::SendInLoop,this,msg));
        }
        void Send(const string& data,int size){
            string msg(data,size);
            _loop->PushTask(std::bind(&Connection::SendInLoop,this,std::move(msg)));
        }
        //关闭连接
        void Close() {
            _loop->PushTask(std::bind(&Connection::HandleCloseEvent,this));
        }
        //判断是否连接成功
        bool IsConnected(){
            return _status==CONNECTED;
        }
        //获取事件循环
        EventLoop* GetEventLoop(){
            return _loop;
        }
        const Any& GetAny() const{
            return _any;
        }
        void SetContext(const Any& any){
            _any=any;
        }
        void Upgrade(const Any &context, const ConnectedCallback &conn, const MessageCallback &msg, 
                     const ClosedCallback &closed, const AnyEventCallback &event) {
            _loop->PushTask(std::bind(&Connection::UpgradeInLoop,this,context,conn,msg,closed,event));
        }
        
};
class Accept{
    private:
        //注意初始化列表顺序
        Socket _socket;
        EventLoop* _loop;
        Channel _channel;
        using AcceptCallback=std::function<void(int)>;
        AcceptCallback _accpet_callback;
        void HandleRead(Channel* channel){
            LOG(ly::LogLevel::DEBUG)<<"accept callback";
            int fd=channel->Fd();
            int clientfd=accept(fd,NULL,NULL);
            if(clientfd<0){
                perror("accept error");
                return;
            }
            _accpet_callback(clientfd);
            LOG(ly::LogLevel::DEBUG)<<"accept client fd:"<<clientfd;
        }
        int CreateServer(const std::string& ip,uint16_t port){
            _socket.CreateServer(ip,port);
            if(_socket.Fd()<0){
                perror("create server error");
                std::cout<<__LINE__<<std::endl;
                exit(1);
            }
            return _socket.Fd();
        }
    public:
        Accept(EventLoop* loop,int port):_socket(CreateServer("0.0.0.0",port)),_loop(loop),_channel(_socket.Fd(),loop){
            LOG(ly::LogLevel::DEBUG)<<"Accept创建"<<" "<<_socket.Fd()<<" "<<port;
            _channel.SetReadCallback(std::bind(&Accept::HandleRead,this,&_channel));
        }
        void SetAcceptCallback(const AcceptCallback& callback){
            LOG(ly::LogLevel::DEBUG)<<"SetAcceptCallback";
            _accpet_callback=callback;
        }
        //必须要等构造函数执行完毕后再调用
        void Listen(){
            LOG(ly::LogLevel::DEBUG)<<"Listen"<<" "<<_channel.Fd();
            _channel.EnableReadEvent();
        }
        int Fd(){
            return _socket.Fd();
        }
};
class LoopThread{
    private:
        std::mutex _mutex;
        std::condition_variable _cond;
        EventLoop* _loop;
        std::thread _thread;
        void ThreadEntry(){
            EventLoop loop(std::this_thread::get_id());
            {
                std::lock_guard<std::mutex> lock(_mutex);
                _loop=&loop;
                _cond.notify_all();
            }
            _loop->Start();
        }
    public:
        LoopThread():_loop(nullptr),_thread(&LoopThread::ThreadEntry,this){}
        EventLoop* GetLoop(){
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock,[&](){return _loop!=nullptr;});
            if(_loop==nullptr){
                assert(0);
            }
            return _loop;
        }

};
class ThreadPool{
    private:
        std::vector<LoopThread*> _threads;
        std::vector<EventLoop*> _loops;
        int _thread_num,idx;
    public:
        ThreadPool(int thread_num):_thread_num(thread_num),idx(0){
        }
        void SetThreadNum(int thread_num){
            _thread_num=thread_num;
        }
        void Create(){
            if(_thread_num<=0){
                return;
            }
            _threads.resize(_thread_num);
            _loops.resize(_thread_num);
            for(int i=0;i<_thread_num;i++){
                _threads[i]=new LoopThread();
                _loops[i]=_threads[i]->GetLoop();
            }
        }
        EventLoop* NextLoop(){
            if(idx<0){
                assert(0);
            }
            if(idx>=_thread_num){
                idx%=_thread_num;
            }
            return _loops[idx++];
        }
};
class TcpServer{
    private:
        using callback_func=std::function<void(void)>;
        using ConnectedCallback = std::function<void( Connection*)>;//连接建立完成后应该干什么
        using MessageCallback = std::function<void( Connection*, Buffer *)>;//收到消息后应该干什么
        using ClosedCallback = std::function<void( Connection*)>;//连接关闭后应该干什么
        using AnyEventCallback = std::function<void( Connection*)>;//任何事件发生时应该干什么
        bool _enable_inactive_release;
        int _timeout,_port,_thread_cnt,_conn_id;
        EventLoop* _main_loop;
        Accept _accept;
        ThreadPool _thread_pool;
        std::unordered_map<int,Connection*> _connections;

        ConnectedCallback _connected_callback;
        MessageCallback _message_callback;
        ClosedCallback _closed_callback;
        AnyEventCallback _event_callback;
        void HandleClose(Connection* conn){
            int id=conn->Id();
            assert(_connections[id]);
            delete _connections[id];
            _connections.erase(id);
        }
        void CreateConnection(int fd){
            _conn_id++;
            EventLoop* loop=nullptr;
            if(_thread_cnt<=0){
                loop=_main_loop;
            }
            else{
                loop=_thread_pool.NextLoop();
            }
            if(loop==nullptr){
                assert(0);
            }
            LOG(ly::LogLevel::DEBUG)<<"CreateConnection:"<<fd;
            Connection* conn=new Connection(loop,_conn_id,fd);
            _connections[_conn_id]=conn;
            // client->SetConnectedCallback(bind(&CnnectCallback,client));
            // client->SetMessageCallback(&ReadCallback);
            // client->SetCloseCallback(bind(&callback,client));
            // client->SetAnyEventCallback(bind(&EventCallback,client));
            conn->SetConnectedCallback(_connected_callback);
            conn->SetMessageCallback(_message_callback);
            conn->SetCloseCallback(_closed_callback);
            conn->SetAnyEventCallback(_event_callback);
            conn->SetServerClosedCallback(std::bind(&TcpServer::HandleClose,this,conn));
            conn->Established();
            if(_enable_inactive_release){
                conn->EnableInactiveRelease(_timeout);
            }
        }
    public:
        TcpServer(int port,int thread_cnt=1,int timeout=30000):_port(port),_thread_cnt(thread_cnt),_enable_inactive_release(false),_conn_id(0),
                _timeout(timeout),_main_loop(new EventLoop(std::this_thread::get_id())),_accept(_main_loop,port),_thread_pool(thread_cnt){
            _accept.SetAcceptCallback(std::bind(&TcpServer::CreateConnection,this,std::placeholders::_1));
        }
        // client->SetConnectedCallback(bind(&CnnectCallback,client));
        // client->SetMessageCallback(&ReadCallback);
        // client->SetCloseCallback(bind(&callback,client));
        // client->SetAnyEventCallback(bind(&EventCallback,client));
        void SetConnectedCallback(const ConnectedCallback& callback){
            _connected_callback=callback;
        }
        void SetMessageCallback(const MessageCallback& callback){
            _message_callback=callback;
        }
        void SetCloseCallback(const ClosedCallback& callback){
            _closed_callback=callback;
        }
        void SetAnyEventCallback(const AnyEventCallback& callback){
            _event_callback=callback;
        }
        void AddTimerTask(int id,int timeout,const callback_func& callback){
            _main_loop->AddTimerTask(id,timeout,callback);
        }
        void EnableInactiveRelease(int timeout) { 
            _timeout = timeout; _enable_inactive_release = true; 
        }
        void SetThreadcnt(int cnt){
            _thread_cnt=cnt;
        }
        void Start(){
            _thread_pool.Create();
            _accept.Listen();
            _main_loop->Start();
        }
};